package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeProvinceOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.MyClickhouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2023/1/11
 * 交易域：省份粒度下单业务过程聚合统计
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、clickhouse
 *      DwdTradeOrderDetail、DwsTradeProvinceOrderWindow
 */
public class DwsTradeProvinceOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        /*
        //TODO 2.检查点相关的设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消之后  检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop202:8020/gmall/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */
        //TODO 3.从kafka的下单主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_province_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.过滤掉流中null消息以及对流中数据类型进行转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                if (jsonStr != null) {
                    JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                    out.collect(jsonObj);
                }
            }
        });
        // {"create_time":"2022-12-27 09:39:31","sku_num":"3","activity_rule_id":"1",
        // "split_original_amount":"12564.0000","sku_id":"13","source_type":"2402","date_id":"2022-12-27",
        // "source_type_name":"商品推广","user_id":"231","province_id":"32","activity_id":"1",
        // "sku_name":"华为 HUAWEI ","id":"1333","source_id":"82","order_id":"544","split_activity_amount":"500.0",
        // "split_total_amount":"12064.0","ts":"1673401171"}
        // jsonObjDS.print(">>>>");
        //TODO 5.去重
        //5.1 按照唯一键进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS
            = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //5.2 使用状态编程  + 度量值的抵销完成去重
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastValueState", JSONObject.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(30)).build());
                    lastValueState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastValue = lastValueState.value();
                    if (lastValue != null) {
                        //说明有重复数据  将状态中的jsonObj中影响度量值的属性取反
                        String splitTotalAmount = lastValue.getString("split_total_amount");
                        lastValue.put("split_total_amount", "-" + splitTotalAmount);
                        out.collect(lastValue);
                    }
                    out.collect(jsonObj);
                    lastValueState.update(jsonObj);
                }
            }
        );
        //TODO 6.对流中数据类型进行转换    jsonObj->实体类对象
        SingleOutputStreamOperator<TradeProvinceOrderBean> orderBeanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeProvinceOrderBean>() {
                @Override
                public TradeProvinceOrderBean map(JSONObject jsonObj) throws Exception {
                    String provinceId = jsonObj.getString("province_id");
                    String orderId = jsonObj.getString("order_id");
                    String splitTotalAmount = jsonObj.getString("split_total_amount");
                    Long ts = jsonObj.getLong("ts") * 1000;
                    TradeProvinceOrderBean orderBean = TradeProvinceOrderBean.builder()
                        .provinceId(provinceId)
                        .orderIdSet(new HashSet<>(Collections.singleton(orderId)))
                        .orderAmount(new BigDecimal(splitTotalAmount))
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );
        // orderBeanDS.print(">>");
        //TODO 7.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeProvinceOrderBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeProvinceOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );
        //TODO 8.按照省份id进行分组
        KeyedStream<TradeProvinceOrderBean, String> provinceIdKeyedDS = withWatermarkDS.keyBy(TradeProvinceOrderBean::getProvinceId);
        //TODO 9.开窗
        WindowedStream<TradeProvinceOrderBean, String, TimeWindow> windowDS
            = provinceIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //TODO 10.聚合统计
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeProvinceOrderBean>() {
                @Override
                public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1, TradeProvinceOrderBean value2) throws Exception {
                    value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                    return value1;
                }
            },
            new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String groupId, TimeWindow window, Iterable<TradeProvinceOrderBean> input, Collector<TradeProvinceOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    for (TradeProvinceOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setTs(System.currentTimeMillis());
                        orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                        out.collect(orderBean);
                    }
                }
            });
        // reduceDS.print(">>>");
        //TODO 11.关联省份维度
        SingleOutputStreamOperator<TradeProvinceOrderBean> withProvinceDS = AsyncDataStream.unorderedWait(
            reduceDS,
            new DimAsyncFunction<TradeProvinceOrderBean>("dim_base_province") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeProvinceOrderBean orderBean) {
                    orderBean.setProvinceName(dimInfoJsonObj.getString("NAME"));
                }

                @Override
                public String getKey(TradeProvinceOrderBean orderBean) {
                    return orderBean.getProvinceId();
                }
            },
            60, TimeUnit.SECONDS
        );

        //TODO 12.将关联的结果写到Clickhouse中
        withProvinceDS.print(">>>");
        withProvinceDS.addSink(
            MyClickhouseUtil.getSinkFunction("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?)")
        );
        env.execute();
    }
}
